package com.heima.schedule.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.heima.clients.schedule.ITaskClient;
import com.heima.common.redis.CacheService;
import com.heima.model.common.dtos.ResponseResult;
import com.heima.model.common.enums.AppHttpCodeEnum;
import com.heima.model.schedule.dto.TaskDto;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.mapper.TaskInfoLogsMapper;
import com.heima.schedule.mapper.TaskInfoMapper;
import com.heima.schedule.service.TaskInfoService;
import org.apache.avro.data.Json;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.List;
import java.util.Set;

/**
 * @author xiaquanen
 * @version 1.0
 * @project heima-leadnews
 * @date 2024/7/26 09:57:13
 */
@Service
public class TaskInfoServiceImpl implements TaskInfoService {

    @Autowired
    private TaskInfoMapper taskInfoMapper;

    @Autowired
    private TaskInfoLogsMapper taskInfoLogsMapper;

    @Autowired
    private CacheService cacheService;

    /**
     * 添加任务
     *
     * @param taskDto
     * @return
     */
    @Override
    public ResponseResult addTask(TaskDto taskDto) {
        //先将数据添加到数据库中
        addTaskDb(taskDto);
        //将数据库中的数据同步到redis中去，根据时间存放到redis中去
        addTaskRedis(taskDto);
        return null;
    }

    /**
     * 根据时间存放到不同的数据库中去
     *
     * @param taskDto
     */
    private void addTaskRedis(TaskDto taskDto) {
        //获取发布时间
        long executeTime = taskDto.getExecuteTime();
        //获取当前时间
        long nowTime = System.currentTimeMillis();
        //获取未来时间
        long futureTime = nowTime+(1000*60*5);
        //根据时间进行判断，存储到哪个集合中去
        if(executeTime<=nowTime)
        {
            //添加到list集合中去
            cacheService.lLeftPush("topic_"+taskDto.getTaskType()+"_"+taskDto.getPriority(),JSON.toJSONString(taskDto));
        }else if(executeTime<=futureTime)
        {
            cacheService.zAdd("future_"+taskDto.getTaskType()+"_"+taskDto.getPriority(),JSON.toJSONString(taskDto),executeTime);
        }
    }

    /**
     * 添加任务到数据库中去
     *
     * @param taskDto
     */
    private void addTaskDb(TaskDto taskDto) {
        //创建一个数据库的实体类
        Taskinfo taskinfo = new Taskinfo();
        BeanUtils.copyProperties(taskDto, taskinfo);
        //实体类中的时间类型不一致需要重新赋值
        taskinfo.setExecuteTime(new Date(taskDto.getExecuteTime()));
        //添加到数据库中去
        taskInfoMapper.insert(taskinfo);
        //将mybatisPlus中生成的id返回给dto中去，方便以后redis中消费者拿完数据后进行删除操作
        taskDto.setTaskId(taskinfo.getTaskId());

        //再添加数据日志的数据库
        TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
        BeanUtils.copyProperties(taskinfo,taskinfoLogs);
        //将日志中的乐观锁进行和状态进行单独赋值
        taskinfoLogs.setStatus(0);
        taskinfoLogs.setVersion(1);
        //进行持久化
        taskInfoLogsMapper.insert(taskinfoLogs);
    }

    /**
     * 拉取任务
     *
     * @param taskType
     * @param priority
     * @return
     */
    @Override
    public ResponseResult pullTask(Integer taskType, Integer priority) {
        //拼接出redis中的key
        String key="topic_"+taskType+"_"+priority;
        String taskJSON = cacheService.lRightPop(key);
        //判断taskJSON是否为空
        if(StringUtils.isNotEmpty(taskJSON))
        {
            TaskDto taskDto = JSON.parseObject(taskJSON, TaskDto.class);
            //删除taskinfo表中的数据
            taskInfoMapper.deleteById(taskDto.getTaskId());
            //修改taskinfologs中的状态
            taskInfoLogsMapper.update(null, Wrappers.<TaskinfoLogs>lambdaUpdate().eq(TaskinfoLogs::getTaskId,taskDto.getTaskId()).set(TaskinfoLogs::getStatus,1));
            //直接将JSON格式的字符串返回
            return ResponseResult.okResult(taskJSON);
        }
        //队列中不存在数据返回非200的状态码
        return ResponseResult.errorResult(AppHttpCodeEnum.DATA_NOT_EXIST);
    }

    /**
     * 每隔一分钟将ZSet中的消息刷新到list中去
     *
     */
    @Scheduled(cron = "0 0/1 * * * ?")
    public void refreshZSetToList()
    {
        //查询所有future相关的key
        Set<String> keys = cacheService.scan("future_*");
        if(keys!=null&& !keys.isEmpty())
        {
            for (String key : keys) {
                //获取集合中的元素时间（分数）满足要求的
                Set<String> tackJSONS = cacheService.zRangeByScore(key, 0, System.currentTimeMillis());
                if(tackJSONS!=null&& !tackJSONS.isEmpty())
                {
                   //生成list中的key的名称
                    String listKey="topic_"+key.split("future_")[1];
                    //批量同步到list中去
                    cacheService.refreshWithPipeline(key, listKey,tackJSONS);
                }
            }
        }
    }

    /**
     * 每隔五分钟将数据组中的消息刷新到redis中去
     *
     */
    @Scheduled(cron = "0 0/5 * * * ?")
    public void refreshDBToRedis()
    {
        //查询数据库中的数据，时间满足的就刷新到redis中去
        //计算期望时间
        long futureTime = System.currentTimeMillis() + (1000 * 60 * 5);
        Date date = new Date(futureTime);
        List<Taskinfo> taskinfos = taskInfoMapper.selectList(Wrappers.<Taskinfo>lambdaQuery().le(Taskinfo::getExecuteTime, date));
        if(taskinfos!=null&& !taskinfos.isEmpty())
        {
            for (Taskinfo taskinfo : taskinfos) {
                TaskDto taskDto = new TaskDto();
                BeanUtils.copyProperties(taskinfo,taskDto);
                taskDto.setExecuteTime(taskinfo.getExecuteTime().getTime());
                addTaskRedis(taskDto);
            }
        }
    }
}
